package com.slipper.core.message;

import com.slipper.util.Tool;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;

import java.util.*;
import java.util.concurrent.*;

public class Publisher {
    private final Logger logger = LoggerFactory.getLogger(Publisher.class);
    private final Set<Subscriber> subscriberList = new HashSet<>();
    private final BlockingQueue<Message> queue = new LinkedBlockingQueue<>();
    private final Object lock = new Object();
    private boolean shutdown = false;
    private boolean isMonitor = false;
    private Timer timer;

    public Publisher() {
        createConsumer();
    }

    public void subscribe(String topic, Subscriber subscriber) {
        if (Tool.isBlank(topic) || subscriber == null || subscriberList.contains(subscriber)) return;

        synchronized(this.lock) {
            this.subscriberList.add(subscriber);
        }

        startMonitor();

        if (Topic.LOGGER.equals(topic)) {
            MessageUtil.setSubscribeLogger(true);
        }
    }

    public void unsubscribe(Subscriber subscriber) {
        if (subscriber == null) return;

        synchronized(this.lock) {
            this.subscriberList.remove(subscriber);
        }

        if (this.subscriberList.stream().noneMatch(item -> item.isTopic(Topic.LOGGER))) {
            MessageUtil.setSubscribeLogger(false);
        }

        subscriber.update(Message.CLOSE_MESSAGE);
    }

    public void publish(String topic, String message) {
        if (this.shutdown || Tool.isBlank(topic) || this.subscriberList.isEmpty()) return;

        try {
            queue.put(new Message(topic, message));
        } catch (InterruptedException e) {
            logger.error("消息发布被打断！");
        } catch (Exception e) {
            logger.error("消息发布出错！", e);
        }
    }

    public void shutdown() {
        this.shutdown = true;
    }

    private void createConsumer() {
        Runnable consumer = () -> {
            while (!this.shutdown) {
                try {
                    Message message = this.queue.take();
                    if (this.subscriberList.isEmpty()) continue;

                    this.subscriberList.stream()
                            .filter(subscriber -> subscriber != null && subscriber.isTopic(message.getTopic()))
                            .forEach(subscriber -> subscriber.update(message));
                } catch (InterruptedException e) {
                    logger.error("消息发送被打断！");
                } catch (Exception e) {
                    logger.error("消息发送出错！", e);
                }
            }
        };
        new Thread(consumer).start();
    }

    private void startMonitor() {
        if (this.isMonitor) return;

        synchronized(lock) {
            if (this.isMonitor) return;

            this.isMonitor = true;
            this.timer = new Timer();
            this.timer.schedule(new TimerTask() {
                public void run() {
                    logger.info("当前消息订阅数：{}", subscriberList.size());
                    if (subscriberList.isEmpty()) {
                        stopMonitor();
                    }
                }
            }, 0, 30 * 60 * 1000L);
        }
    }

    private void stopMonitor() {
        synchronized(lock) {
            this.isMonitor = false;
            this.timer.cancel();
            this.timer = null;
        }
    }
}
